package cn.cs.message.consumer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import cn.cs.message.Message;
import cn.cs.message.ReconnectHandler;

public abstract class MessageConsumerSerial extends MessageConsumer
{
	private static final Logger logger=LoggerFactory.getLogger(MessageConsumerSerial.class);
	
	protected Map<Integer,Channel> channels=new HashMap<Integer, Channel>();
	protected Map<Integer,DefaultConsumer> consumers=new HashMap<Integer, DefaultConsumer>();
	
	private void makeConsumer(int index)
	{
		while(true)
		{
			Channel channel=null;
			DefaultConsumer consumer=null;
			try{
				channel = channels.get(index);
				if(channel!=null)
				{
					if(channel.isOpen())
						channel.close();
					channels.remove(index);
				}
				channel=connectManage.createChannel();
				if(channel!=null)
				{	
					channel.basicQos(qos);
					channels.put(index, channel);
				}	
				consumer=work(index);
				if(consumer!=null)
				{
					consumers.put(index, consumer);
					break;
				}	
			}catch(Exception e){
				logger.warn("等待1分钟后再尝试建立consumer和队列的连接",e);
			}
			if(channel==null||consumer==null)
				try {
					TimeUnit.SECONDS.sleep(60);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
		}
	}
	
	public DefaultConsumer work(final int index) throws IOException 
	{
		final Channel channel=channels.get(index);
		if(channel==null)
			return null;
		boolean autoAck=false;
		DefaultConsumer consumer =  new DefaultConsumer(channel) {
			@Override
		    public void handleCancel(String consumerTag) throws IOException {
				reminder.remind("消费的队列可能丢失");
				makeConsumer(index);
			}
	         @Override
	         public void handleDelivery(String consumerTag,
	                                    Envelope envelope,
	                                    AMQP.BasicProperties properties,
	                                    byte[] body) throws IOException
	             {
	        	 	long deliveryTag = envelope.getDeliveryTag();
	        	 	boolean suc=true;
	        	 	try {
	        	 		Message msg=getFromBytes(body);
	        	 		handleMessage(msg,envelope.isRedeliver());
					} catch (Exception e){
						logger.error("未预见的错误",e);
						suc=false;
					} finally{
						if(suc)
							channel.basicAck(deliveryTag, false);
						else
							channel.basicNack(deliveryTag, false,true);
					}
	             }
	         };	    	
    	channel.basicConsume(messageNameService.getQueueName(appCode), autoAck, consumer);		    	
    	return consumer;
	}
	
	/**消息处理
	 * @param msg 消息实体*/
	public abstract void handleMessage(Message msg,boolean isRedeliver);
	
	@Override
	protected void doStart()
	{
		reconnectHandler.afterReconnect();
	}
	
	@Override
	protected void doStop() throws IOException, TimeoutException
	{
		for(Channel channel:channels.values())
			channel.close();
	}
	
	@Override
	public ReconnectHandler getReconnectHandler() {
		if(reconnectHandler==null)
		{
			reconnectHandler=new ReconnectHandler(){
				@Override
				public void afterReconnect() {
					for(int i=0;i<threadPoolSize;i++)
						makeConsumer(i);
				}
			};
		}
		return reconnectHandler;
	}
}
